package com.iphanta.community.mq;

import com.baidu.fsg.uid.UidGenerator;
import com.iphanta.community.common.utils.RedisKey;
import com.iphanta.community.common.utils.RedisKeyUtils;
import com.iphanta.community.following.entity.Feed;
import com.iphanta.community.user.entity.Relation;
import com.iphanta.community.following.mapper.FeedMapper;
import com.iphanta.community.user.mapper.RelationMapper;
import com.iphanta.community.mq.message.DynamicFeed;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * Feed流消费端
 *
 * @author 墨声 E-mail: zehong.hongframe.huang@gmail.com
 * create time: 2/18/22 6:34 PM
 */
@Service
@RocketMQMessageListener(
        topic = "dynamic-publish",
        consumerGroup = "dynamic-producer-group",
        consumeThreadMax = 32
)
public class DynamicFeedListener implements RocketMQListener<DynamicFeed> {

    private final static Logger LOGGER = LoggerFactory.getLogger(DynamicFeedListener.class);

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private RelationMapper relationMapper;

    @Autowired
    private FeedMapper feedMapper;

    @Autowired
    private UidGenerator uidGenerator;

    private final static String LPUSH_IF_PRESENT_SCRIPT =
            "if redis.call('exists',KEYS[1]) == 1 then\n" +
                    " redis.call('lpush', KEYS[1], ARGV[1])\n" +
                    "return 1\n" +
                    "end";

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void onMessage(DynamicFeed message) {
        MDC.put("requestId", message.getRequestId());
        try {
            String self = RedisKeyUtils.INSTANCE.key(RedisKey.LIST_FEED, message.getAuthorId());
            lpushIfPresent(self, String.valueOf(message.getDynamicId()));
            //select fans query in relation table
            List<Relation> fans = this.relationMapper.selectFansList(message.getAuthorId());
            if (CollectionUtils.isEmpty(fans)) {
                LOGGER.info("用户{}暂无粉丝", message.getAuthorId());
                return;
            }
            List<String> fansIds = fans.stream()
                    .map(e -> RedisKeyUtils.INSTANCE.key(RedisKey.VALUE_ACTIVE, e.getFollowerId()))
                    .collect(Collectors.toList());
            //foreach query of active fans (include online), then insert mysql
            List<String> active = redisTemplate.opsForValue().multiGet(fansIds).stream()
                    .filter(e -> Objects.nonNull(e)).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(active)) {
                LOGGER.info("{} 粉丝列表暂无活跃用户", message.getAuthorId());
                return;
            }
            List<Feed> feeds = active.stream().map(e -> {
                Feed feed = new Feed();
                feed.setId(uidGenerator.getUID());
                feed.setDynamicId(message.getDynamicId());
                feed.setFollowingId(message.getAuthorId());
                feed.setFollowerId(e);
                feed.setDynamicCreateTime(message.getCreateTime());
                return feed;
            }).collect(Collectors.toList());
            this.feedMapper.batchInsert(feeds);
            LOGGER.info("用户 {} 的活跃粉丝数 : {}, ID如下\n {}\n", message.getAuthorId(), active.size(), active);

            //foreach query of online fans, then set redis
            List<String> online = new ArrayList<>();
            for (String followerId : active) {
                String key = RedisKeyUtils.INSTANCE.key(RedisKey.LIST_FEED, followerId);
                Long isSuccess = lpushIfPresent(key, String.valueOf(message.getDynamicId()));
                if (Objects.equals(isSuccess, 1L)) {
                    online.add(followerId);
                }
            }
            LOGGER.info("用户 {} 的在线粉丝数 : {}. ID如下\n {}\n", message.getAuthorId(), online.size(), online);
        } finally {
            MDC.remove("requestId");
        }
    }

    private Long lpushIfPresent(String key, String value) {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Long.class);
        redisScript.setScriptText(LPUSH_IF_PRESENT_SCRIPT);
        return this.redisTemplate.execute(redisScript, Arrays.asList(key), value);
    }
}
